#pragma once
#ifndef __MY_SERVER__
#define __MY_SERVER__

#include "../mqcommon/timer.hpp"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/TcpServer.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "./connection.hpp"

namespace NS_Broker
{
    class Broker
    {
        const std::string host_name = "virtual_host";
        const std::string basedir = "./data";
        const std::string dbfilepath = basedir + "/meta";
        // const int thread_num = 2;
        const uint64_t survival_time = 10;

        using ConnectionTimestampPtr = std::shared_ptr<NS_Protocol::ConnectionTimestamp>;

    public:
        // Server端收到的肯定是请求报文，所以我们需要将我们定义好的序列化字段的类型定义一个指针，因为muduo库中使用的都是智能指针

        Broker(int port)
            : _dispatcher(std::bind(&Broker::defaultMessageCallback, this, std::placeholders::_1,
                                    std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1,
                                                               std::placeholders::_2, std::placeholders::_3))),
              _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
              _connection_manager(std::make_shared<NS_Connection::ConnectionManager>()),
              _virtual_host(std::make_shared<NS_Host::VirtualHost>(host_name, basedir, dbfilepath)),
              _consumer_manager(std::make_shared<NS_Consumer::ConsumerManager>()),
              _thread_pool(std::make_shared<ThreadPool>()),
              _timer(std::make_shared<Timer>())
        {
            // 1.如果在数据恢复中有消息队列的话，我们同时也要建立消费者的订阅队列
            // 获取消息队列Map
            // 创建消费者队列单元
            const NS_MsgQueue::MsgQueueMap &msgqueue_map = _virtual_host->getMsgQueues();
            for (auto &[qname, ptr] : msgqueue_map)
            {
                _consumer_manager->initQueueConsumer(qname);
            }

            // 2. 将信道与连接提供的服务绑定给dispatcher中
            _dispatcher.registerMessageCallback<NS_Protocol::OpenChannelRequest>(std::bind(&Broker::openChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::CloseChannelRequest>(std::bind(&Broker::closeChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _dispatcher.registerMessageCallback<NS_Protocol::DeclareExchangeRequest>(std::bind(&Broker::declareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::DeleteExchangeRequest>(std::bind(&Broker::deleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _dispatcher.registerMessageCallback<NS_Protocol::DeclareMsgQueueRequest>(std::bind(&Broker::declareMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::DeleteMsgQueueRequest>(std::bind(&Broker::deleteMsgQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _dispatcher.registerMessageCallback<NS_Protocol::CreateBindingRequest>(std::bind(&Broker::bind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::DeleteBindingRequest>(std::bind(&Broker::unBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _dispatcher.registerMessageCallback<NS_Protocol::CreateConsumerRequest>(std::bind(&Broker::basicConsume, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::DeleteConsumerRequest>(std::bind(&Broker::basicCancel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _dispatcher.registerMessageCallback<NS_Protocol::BasicPublish>(std::bind(&Broker::basicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::BasicAck>(std::bind(&Broker::basicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _dispatcher.registerMessageCallback<NS_Protocol::ConnectionTimestamp>(std::bind(&Broker::updateConnectionTime, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _server.setConnectionCallback(std::bind(&Broker::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }

        void start()
        {
            _server.start();
            _loop.loop();
        }

        ~Broker() {}

    private:
        void timeOut(const muduo::net::TcpConnectionPtr &conn)
        {
            if (_survival.find(conn) == _survival.end())
            {
                DLOG("没有找到对应的连接定时器，数据错误");
                exit(0);
            }

            if (_timer->now() - _survival[conn].second.load() > survival_time)
            {
                DLOG("连接超时，当前时间戳：%ld，连接最近发送的时间戳：%ld", _timer->now(), _survival[conn].second.load());
                _timer->remove(_survival[conn].first);
                _connection_manager->delConnection(conn);
                conn->shutdown();
            }
            else
            {
                DLOG("连接未超时");
            }
        }

        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                DLOG("新连接到来");
                _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _thread_pool);
                DLOG("生成uuid");
                std::string id = NS_Helper::Uuid::uuid();
                _survival[conn] = std::make_pair(id, _timer->now());
                DLOG("时间任务开始建立");
                _timer->push(std::bind(&Broker::timeOut, this, conn), 0, 5, id);
                DLOG("时间任务建立完成");
            }
            else
            {
                DLOG("关闭连接");
                _connection_manager->delConnection(conn);
            }
        }

        void defaultMessageCallback(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        // 创建信道
        void openChannel(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::OpenChannelRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("创建信道时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (channel != nullptr)
            {
                DLOG("信道重复创建");
                return;
            }
            DLOG("创建信道成功：%s", message->channel_id().c_str());
            _my_conn->openChannel(message);
        }
        // 删除信道
        void closeChannel(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::CloseChannelRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("关闭信道时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            DLOG("删除信道成功: %s", message->channel_id().c_str());
            _my_conn->closeChannel(message);
        }
        // 声明交换机
        void declareExchange(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::DeclareExchangeRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("声明交换机时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("声明交换机时没有找到对应的信道");
                return;
            }
            DLOG("声明交换机成功：%s", message->ename().c_str());
            channel->declareExchange(message);
        }
        // 删除交换机
        void deleteExchange(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::DeleteExchangeRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("删除交换机时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("删除交换机时没有找到对应的信道");
                return;
            }
            DLOG("删除交换机成功: %s", message->ename().c_str());
            channel->deleteExchange(message);
        }
        // 声明消息队列
        void declareMsgQueue(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::DeclareMsgQueueRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("声明消息队列时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("声明消息队列时没有找到对应的信道");
                return;
            }
            DLOG("声明消息队列成功: %s", message->qname().c_str());
            channel->declareMsgQueue(message);
        }
        // 删除消息队列
        void deleteMsgQueue(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::DeleteMsgQueueRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("删除消息队列时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("删除消息队列时没有找到对应的信道");
                return;
            }
            DLOG("删除消息队列成功: %s", message->qname().c_str());
            channel->deleteMsgQueue(message);
        }
        // 指定队列与交换机建立绑定
        void bind(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::CreateBindingRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("指定队列与交换机建立绑定时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("指定队列与交换机建立绑定时没有找到对应的信道");
                return;
            }
            DLOG("指定队列与交换机建立绑定成功: %s - %s - %s", message->ename().c_str(), message->qname().c_str(), message->binding_key().c_str());
            channel->Bind(message);
        }
        // 指定队列与交换机解除绑定
        void unBind(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::DeleteBindingRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("指定队列与交换机解除绑定时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("指定队列与交换机解除绑定时没有找到对应的信道");
                return;
            }
            DLOG("指定队列与交换机解除绑定成功: %s - %s", message->ename().c_str(), message->qname().c_str());
            channel->unBind(message);
        }
        // 消息的发布
        void basicPublish(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::BasicPublishPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("消息发布时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("消息发布时没有找到对应的信道");
                return;
            }

            // DLOG("是否持久化：%d", message->mutable_properties()->mode() == NS_Message::DeliveryMode::DURABLE);
            channel->basicPublic(message);
        }
        // 消息的确认
        void basicAck(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::BasicAckPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("消息确认时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("消息确认时没有找到对应的信道");
                return;
            }
            // DLOG("消息确认成功: %s", message->qname().c_str());
            channel->basicAck(message);
        }
        // 对指定消息队列的订阅
        void basicConsume(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::CreateConsumerRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("对指定消息队列的订阅时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("对指定消息队列的订阅时没有找到对应的信道");
                return;
            }
            DLOG("对指定消息队列的订阅成功: %s", message->qname().c_str());
            channel->queueSubscription(message);
        }
        // 对指定消息队列的取消订阅
        void basicCancel(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::DeleteConsumerRequestPtr &message, muduo::Timestamp)
        {
            NS_Connection::Connection::Ptr _my_conn = _connection_manager->getConnection(conn);
            if (_my_conn == nullptr)
            {
                DLOG("对指定消息队列的取消订阅时没有找到对应的连接");
                conn->shutdown();
                return;
            }
            NS_Channel::Channel::Ptr channel = _my_conn->getChannel(message->channel_id());
            if (_my_conn == nullptr)
            {
                DLOG("对指定消息队列的取消订阅时没有找到对应的信道");
                return;
            }
            DLOG("对指定消息队列的取消订阅成功: %s", message->qname().c_str());
            channel->queueCancelSubscription(message);
        }

        // 更新连接的存活时间
        void updateConnectionTime(const muduo::net::TcpConnectionPtr &conn, const ConnectionTimestampPtr &message, muduo::Timestamp)
        {
            if (_survival.find(conn) == _survival.end())
            {
                DLOG("没有找到对应的连接定时器，数据错误");
                exit(0);
            }

            DLOG("更新连接存活时间，更新前：%ld, 更新后：%ld", _survival[conn].second.load(), message->timestamp());
            _survival[conn].second.store(message->timestamp());
        }

    private:
        // 解析到其中的信息之后，我们需要通过_dispatcher来通过不同的请求类型调用不同的处理函数(这是我们要做的)，最后响应回客户端
        ProtobufDispatcher _dispatcher;
        // 服务启动之后收到的消息都会是基于protobuf结构化字段生成的序列，所以需要使用_codec来解析其中的信息
        NS_Channel::ProtobufCodecPtr _codec;
        // 基本的服务器tcp和事件监听loop
        muduo::net::EventLoop _loop;
        muduo::net::TcpServer _server;

        NS_Connection::ConnectionManager::Ptr _connection_manager;
        NS_Host::VirtualHost::Ptr _virtual_host;
        NS_Consumer::ConsumerManager::Ptr _consumer_manager;
        ThreadPoolPtr _thread_pool;

        Timer::Ptr _timer;
        std::unordered_map<muduo::net::TcpConnectionPtr, std::pair<std::string, std::atomic<uint64_t>>> _survival;
    };
}

#endif